package com.hps.demo.discovery;

import com.hps.demo.netty.client.NettyClient;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author heps
 * @since 2020/8/15
 */
@Slf4j
@Component
public class ConnectManage {

    private NettyClient nettyClient;

    private AtomicInteger roundRobin = new AtomicInteger(0);

    private CopyOnWriteArrayList<Channel> channels = new CopyOnWriteArrayList<>();

    private Map<SocketAddress, Channel> channelNodes = new ConcurrentHashMap<>();

    @Lazy
    @Autowired
    public void setNettyClient(NettyClient nettyClient) {
        this.nettyClient = nettyClient;
    }

    public void updateConnectedServer(List<String> addressList) {
        if (CollectionUtils.isEmpty(addressList)) {
            log.error("没有可用的服务器节点, 全部服务节点已关闭");
            for (Channel channel : channels) {
                SocketAddress remotePeer = channel.remoteAddress();
                Channel handlerNode = channelNodes.get(remotePeer);
                handlerNode.close();
            }
            channels.clear();
            channelNodes.clear();
            return;
        }
        HashSet<SocketAddress> newAllServerNodeSet = new HashSet<>();
        for (String address : addressList) {
            String[] array = StringUtils.split(address, ":");
            if (array.length == 2) {
                newAllServerNodeSet.add(new InetSocketAddress(array[0], Integer.parseInt(array[1])));
            }
        }

        for (SocketAddress serverNodeAddress : newAllServerNodeSet) {
            Channel channel = channelNodes.get(serverNodeAddress);
            if (Objects.nonNull(channel) && channel.isOpen()) {
                log.info("当前服务节点已存在, 无需重新连接, {}", serverNodeAddress);
            } else {
                connectServerNode(serverNodeAddress);
            }
        }

        for (Channel channel : channels) {
            SocketAddress remotePeer = channel.remoteAddress();
            if (!newAllServerNodeSet.contains(remotePeer)) {
                log.info("删除失效服务节点: {}", remotePeer);
                Channel channelNode = channelNodes.get(remotePeer);
                if (Objects.nonNull(channelNode)) {
                    channelNode.close();
                    channels.remove(channel);
                    channelNodes.remove(remotePeer);
                }
            }
        }
    }

    private void connectServerNode(SocketAddress serverNodeAddress) {
        try {
            Channel channel = nettyClient.doConnect(serverNodeAddress);
            addChannel(channel, serverNodeAddress);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void addChannel(Channel channel, SocketAddress serverNodeAddress) {
        log.info("加入channel到连管理器: {}", serverNodeAddress);
        channels.add(channel);
        channelNodes.put(serverNodeAddress, channel);
    }

    public Channel chooseChannel() {
        if (channels.size() > 0) {
            int size = channels.size();
            int index = roundRobin.getAndIncrement() % size;
            return channels.get(index);
        }
        return null;
    }

    public void removeChannel(Channel channel) {
        log.info("从连接管理器中删除失效的channel: {}", channel.remoteAddress());
    }
}
